package com.shujia.stream

import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo7DirectKafka {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("stream")
      .setMaster("local") //指定两个线程

    val sc = new SparkContext(conf)

    //创建spark streaming上下文对象   指定间隔时间
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val params = Map("metadata.broker.list" -> "node1:9092,node2:9092,node3:9092")

    val topics = Set("topic4")

    /**
      * 主动拉去数据
      * 每一个kafka分区对应一个RDD分区
      *
      */
    val kafkaDS = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc,
      params,
      topics
    )


    kafkaDS.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
